package com.atguigu.bigdata.gmall.realtime.app.dwd;

import com.atguigu.bigdata.gmall.realtime.app.BaseSQLApp;
import com.atguigu.bigdata.gmall.realtime.common.Constant;
import com.atguigu.bigdata.gmall.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Dwd_04_CourseDetail extends BaseSQLApp {
    public static void main(String[] args) {
        new Dwd_04_CourseDetail().init(
                4004,
                2,
                "Dwd_04_CourseDetail"
        );
    }
    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        //读取ods_db数据
        readOdsDb(tEnv,"Dwd_04_CourseDetail");

        //tEnv.sqlQuery("select * from ods_db where `table` = 'test_paper'").execute().print();
        //从ods中过滤出test_exam
        Table ExamInfo = tEnv.sqlQuery("select " +
                " `data`['id'] id, " +
                " `data`['paper_id'] paper_id, " +
                " `data`['user_id'] user_id, " +
                " `data`['score'] score, " +
                " `data`['duration_sec'] duration_sec, " +
                " `data`['create_time'] create_time, " +
                " ts," +
                " pt " +
                "from ods_db " +
                "where `database`='gmall2022' " +
                "and `table`='test_exam' " +
                "and `type`='insert' " );
        tEnv.createTemporaryView("test_exam", ExamInfo);

        //ExamInfo.execute().print();
        //从ods中过露出test_exam_question
        Table questionInfo = tEnv.sqlQuery("select " +
                " `data`['id'] id, " +
                " `data`['exam_id'] exam_id, " +
                " `data`['paper_id'] paper_id, " +
                " `data`['question_id'] question_id, " +
                " `data`['user_id'] user_id, " +
                " `data`['is_correct'] is_correct, " +
                " `data`['create_time'] create_time, " +
                " `data`['score'] score, " +
                " ts," +
                "pt  " +
                "from ods_db " +
                "where `database`='gmall2022' " +
                "and `table`='test_exam_question' " +
                "and `type`='insert' " );
        tEnv.createTemporaryView("test_exam_question", questionInfo);
        //questionInfo.execute().print();

        //获取维度
        //readBaseinfo(tEnv);

        Table paperInfo1 = tEnv.sqlQuery("select " +
                " `data`['id'] id, " +
                " `data`['paper_title'] paper_title, " +
                " `data`['course_id'] course_id, " +
                " `data`['create_time'] create_time, " +
                " ts " +
                "from ods_db " +
                "where `database`='gmall2022' " +
                "and `table`='test_paper' " +
                "and `type`='insert' " );
        tEnv.createTemporaryView("test_paper1", paperInfo1);
        //paperInfo.execute().print();


        //三张表join
        Table result = tEnv.sqlQuery("select " +
                "teq.id,  " +
                "tp.paper_title,  " +
                "teq.paper_id, " +
                "tp.course_id, " +
                "teq.score question_score," +
                "teq.question_id, " +
                "te.score exam_score,  " +
                "te.duration_sec,  " +
                "teq.user_id,  " +
                "teq.is_correct,  " +
                "teq.create_time,  "+
                "teq.ts," +
                "current_row_timestamp() row_op_ts    " +
                "from test_exam_question teq " +
                "join test_exam  te " +
                "on teq.exam_id=te.id " +
                "join test_paper1 tp " +
                "on teq.paper_id = tp.id ");

        //result.execute().print();


        //打印会阻塞
        //写出到kafka中
        tEnv.executeSql("create table dwd_exam_with_questionAndId_detail(" +
                "id string,  " +
                "paper_title string,  " +
                "paper_id string,  " +
                "course_id string, " +
                "question_score string," +
                "question_id string,  " +
                "exam_score string,  " +
                "duration_sec string,  " +
                "user_id string,  " +
                "is_correct string,  " +
                "create_time string,  " +
                "ts bigint," +
                "row_op_ts timestamp_ltz(3)," +
                "primary key(id) not enforced " +
                ")" + SQLUtil.getUpsertKafkaSink(Constant.TOPIC_DWD_EXAM_WITH_QUESTIONANDID_DETAIL));

        result.executeInsert("dwd_exam_with_questionAndId_detail");


    }

    public void readBaseinfo(StreamTableEnvironment tEnv) {
        tEnv.executeSql("create table course_info ( " +
                "  id BIGINT, " +
                "  course_name string " +
                ") WITH ( " +
                "  'connector' = 'jdbc', " +
                "  'driver' = 'com.mysql.cj.jdbc.Driver', " +
                "  'url' = 'jdbc:mysql://hadoop162:3306/gmall2022?useSSL=false', " +
                "  'table-name' = 'course_info', " +
                "  'username' = 'root', " +
                "  'password' = 'aaaaaa',  " +
                "  'lookup.cache.max-rows' = '10'," +
                "  'lookup.cache.ttl' = '1 hour' " +
                ")");



    }
}

